package com.lianjia.sh.se.kafka.client.annotion;

import com.lianjia.sh.se.kafka.client.AbstractKafkaListener;
import com.lianjia.sh.se.kafka.client.AbstractKafkaSender;
import com.lianjia.sh.se.kafka.client.config.CustomDefaultListenerConfig;
import com.lianjia.sh.se.kafka.client.constants.ConsumerCons;
import com.lianjia.sh.se.kafka.client.constants.ProducerCons;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import javax.annotation.PostConstruct;
import java.util.Iterator;
import java.util.Map;

/**
 * 描描所有kafka注解
 * Created by yangsp1 on 16/7/18.
 */
@Component
public class KafkaScan implements ApplicationContextAware {
    ApplicationContext applicationContext;

    @Autowired
    CustomDefaultListenerConfig customDefaultListenerConfig;

    @PostConstruct
    public void init(){
        initListeners();
        initSenders();
    }

    /**
     * 初始化producer配置
     */
    private void initListeners(){
        Map<String,AbstractKafkaListener> map = applicationContext.getBeansOfType(AbstractKafkaListener.class);
        Iterator listeners = map.keySet().iterator();

        //获取所有注解配置并启动listener线程
        while(listeners.hasNext()) {
            String key = (String)listeners.next();
            AbstractKafkaListener listener = map.get(key);

            //无注解则忽略
            KafkaListener kafkaListener = listener.getClass().getAnnotation(KafkaListener.class);
            if(kafkaListener == null){
                continue;
            }

            //从注解获取配置
            //topic
            String topic = kafkaListener.topic();
            Assert.isTrue(StringUtils.isNotBlank(topic), "[" + listener.getClass().toString() + "]topic不能为空");
            listener.setTopic(topic);
            //刷新频率：时间
            Long lingerMs = kafkaListener.lingerMs();
            if(lingerMs != null && lingerMs > 0) {
                listener.setLingerMs(lingerMs);
            }else if(customDefaultListenerConfig.getLingerMs() != null && customDefaultListenerConfig.getLingerMs() > 0) {
                listener.setLingerMs(customDefaultListenerConfig.getLingerMs());
            }
            //broker配置
            String bootstrapServers = kafkaListener.bootstrapServers();
            if(StringUtils.isNotBlank(bootstrapServers)){
                listener.getConfigs().put(ConsumerCons.BOOTSTRAP_SERVERS_NAME,bootstrapServers);
            }
            //groupId配置
            String groupId = kafkaListener.groupId();
            if(StringUtils.isNotBlank(groupId)) {
                listener.getConfigs().put(ConsumerCons.GROUP_ID_NAME,groupId);
            }

            //启动listener
            listener.startFrequentPollRecords();
        }
    }

    /**
     * 初始化consumer配置
     */
    private void initSenders(){
        Map<String,AbstractKafkaSender> map = applicationContext.getBeansOfType(AbstractKafkaSender.class);
        Iterator senders = map.keySet().iterator();

        //获取所有注解配置并启动listener线程
        while(senders.hasNext()) {
            String key = (String)senders.next();
            AbstractKafkaSender sender = map.get(key);

            //无注解则忽略
            KafkaSender kafkaSender = sender.getClass().getAnnotation(KafkaSender.class);
            if(senders == null){
                continue;
            }

            //从注解获取配置
            //topic
            String topic = kafkaSender.topic();
            Assert.isTrue(StringUtils.isNotBlank(topic), "[" + sender.getClass().toString() + "]topic不能为空");
            sender.setTopic(topic);
            //刷新频率：时间
            Long lingerMs = kafkaSender.lingerMs();
            if(lingerMs != null && lingerMs > 0) {
                sender.getConfigs().put(ProducerCons.LINGER_MS_NAME,lingerMs);
            }
            //broker配置
            String bootstrapServers = kafkaSender.bootstrapServers();
            if(StringUtils.isNotBlank(bootstrapServers)){
                sender.getConfigs().put(ProducerCons.BOOTSTRAP_SERVERS_NAME,bootstrapServers);
            }
            //client配置
            String clientId = kafkaSender.clientId();
            if(StringUtils.isNotBlank(clientId)) {
                sender.getConfigs().put(ProducerCons.CLIENT_ID_NAME,clientId);
            }
            //batchSize配置
            long batchSize = kafkaSender.batchSize();
            if(batchSize > 0) {
                sender.getConfigs().put(ProducerCons.BATCH_SIZE_NAME,batchSize);
            }

            //requestTimeoutMs
            long requestTimeoutMs = kafkaSender.requestTimeoutMs();
            if(requestTimeoutMs > 0) {
                sender.getConfigs().put(ProducerCons.REQUEST_TIMEOUT_MS_NAME,requestTimeoutMs);
            }

            //acks
            String acks = kafkaSender.acks();
            if(StringUtils.isNotBlank(acks)) {
                sender.getConfigs().put(ProducerCons.ACKS_NAME,acks);
            }

            //retries
            long retries = kafkaSender.retries();
            if(retries > 0) {
                sender.getConfigs().put(ProducerCons.RETRIES_NAME,retries);
            }

            //bufferMemory
            long bufferMemory = kafkaSender.bufferMemory();
            if(bufferMemory > 0) {
                sender.getConfigs().put(ProducerCons.BUFFER_MEMORY_NAME,bufferMemory);
            }
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
